
/*-------------------------------------------------------------------------
 *
 * lwlock.c
 *	  轻量级锁管理器
 *
 * 轻量级锁主要旨在提供对共享内存数据结构的互斥访问。因此，它们提供了独占和共享锁模式（以支持对共享对象的读/写和只读访问）。除了这些之外，没有其他的附加功能。用户级别的锁定应当使用完整的锁管理器——这依赖于LWLocks保护共享状态。
 *
 * 除了独占和共享模式，轻量级锁还可以用来等待变量值的变化。当通过LWLockAcquire获取锁时，变量最初未设置，即保持为上次释放锁时设置的值，并且可以通过调用LWLockUpdateVar在不释放锁的情况下进行更新。LWLockWaitForVar等待变量被更新或锁变为可用。当通过LWLockReleaseClearVar()释放锁时，可以将值设置为一个适合于自由锁的适当值。变量的意义取决于调用者，轻量级锁代码仅仅负责分配和比较它。
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * IDENTIFICATION
 *	  src/backend/storage/lmgr/lwlock.c
 *
 * NOTES:
 *
 * 这曾经是一个相当直接的读写锁实现，其中内部状态由自旋锁保护。不幸的是，获取自旋锁的开销对于在共享模式下非常频繁占用的工作负载/锁来说太高了。我们常常在（显然是独占的）自旋锁中自旋，同时试图获取一个实际上是免费的共享锁。
 *
 * 因此，设计了一种新实现，提供无等待的共享锁获取，适用于未被独占锁定的锁。
 *
 * 基本思想是使用单个原子变量'lockcount'替代以前分开的共享和独占计数器，并利用原子操作来获取锁。这对于普通的rw-spinlocks来说相当简单，但对于希望在操作系统中等待的LWLocks等情况，则困难得多。
 *
 * 对于锁的获取，我们对lockcount变量使用原子比较和交换。对于独占锁，我们用一个哨兵值（LW_VAL_EXCLUSIVE）进行交换，对于共享锁，我们计算持有者的数量。
 *
 * 释放锁时，我们使用原子递减来释放锁。如果新值为零（我们以原子方式获得这个值），我们知道我们可以/必须释放等待者。
 *
 * 显然，确保独占锁的哨兵值与可能的最大共享锁定数量不冲突是非常重要的——幸运的是，MAX_BACKENDS使得这很容易实现。
 *
 * 细心的读者可能注意到，天真的执行上述操作会导致一个明显的竞争条件：我们尝试使用原子操作进行锁定，并注意到我们必须等待。不幸的是，当我们排队完成时，先前的锁定者可能已经完成了它的工作。这是一个问题，因为我们现在被困在操作系统中等待。
 *
 * 为了缓解这些竞争，我们采用两阶段尝试锁定：
 *	 阶段 1：尝试以原子方式执行，如果成功，好。
 *	 阶段 2：将自己添加到锁的等待队列中。
 *	 阶段 3：再尝试获取锁，如果成功，从队列中移除自己。
 *	 阶段 4：睡眠直到唤醒，返回阶段 1。
 *
 * 这可以保护我们免受上述问题，因为在我们排队之前没有人可以释放得太快，因为在阶段 2 之后，我们已经排队。
 * -------------------------------------------------------------------------
 */
#include "postgres.h"

#include "miscadmin.h"
#include "pg_trace.h"
#include "pgstat.h"
#include "port/pg_bitutils.h"
#include "postmaster/postmaster.h"
#include "replication/slot.h"
#include "storage/ipc.h"
#include "storage/predicate.h"
#include "storage/proc.h"
#include "storage/proclist.h"
#include "storage/spin.h"
#include "utils/memutils.h"

#ifdef LWLOCK_STATS
#include "utils/hsearch.h"
#endif


/* 我们使用ShmemLock自旋锁来保护LWLockCounter */
extern slock_t *ShmemLock;

#define LW_FLAG_HAS_WAITERS			((uint32) 1 << 30)
#define LW_FLAG_RELEASE_OK			((uint32) 1 << 29)
#define LW_FLAG_LOCKED				((uint32) 1 << 28)

#define LW_VAL_EXCLUSIVE			((uint32) 1 << 24)
#define LW_VAL_SHARED				1

#define LW_LOCK_MASK				((uint32) ((1 << 25)-1))
/* 必须大于MAX_BACKENDS - 也就是2^23-1，因此我们没问题。 */
#define LW_SHARED_MASK				((uint32) ((1 << 24)-1))

/*
 * LWLock有三种“批次”：
 *
 * 1. 在lwlocknames.h中定义的每个单独命名的锁都有自己的批次。
 * 这些批次的名称出现在lwlocknames.c中的IndividualLWLockNames[]中。
 *
 * 2. 还有一些为内置锁组预定义的批次。
 * 这些在lwlock.h中的枚举BuiltinTrancheIds中列出，它们的名称出现在下面的BuiltinTrancheNames[]中。
 *
 * 3. 扩展可以通过RequestNamedLWLockTranche或LWLockRegisterTranche创建新的批次。
 * 当前进程中已知的这些名称出现在LWLockTrancheNames[]中。
 *
 * 所有这些名称都作为等待事件名称向用户可见，因此要谨慎选择
 * ...并不要忘记更新文档中的等待事件列表。
 */
extern const char *const IndividualLWLockNames[];	/* 在lwlocknames.c中 */

static const char *const BuiltinTrancheNames[] = {
	/* LWTRANCHE_XACT_BUFFER: */
	"XactBuffer",
	/* LWTRANCHE_COMMITTS_BUFFER: */
	"CommitTsBuffer",
	/* LWTRANCHE_SUBTRANS_BUFFER: */
	"SubtransBuffer",
	/* LWTRANCHE_MULTIXACTOFFSET_BUFFER: */
	"MultiXactOffsetBuffer",
	/* LWTRANCHE_MULTIXACTMEMBER_BUFFER: */
	"MultiXactMemberBuffer",
	/* LWTRANCHE_NOTIFY_BUFFER: */
	"NotifyBuffer",
	/* LWTRANCHE_SERIAL_BUFFER: */
	"SerialBuffer",
	/* LWTRANCHE_WAL_INSERT: */
	"WALInsert",
	/* LWTRANCHE_BUFFER_CONTENT: */
	"BufferContent",
	/* LWTRANCHE_REPLICATION_ORIGIN_STATE: */
	"ReplicationOriginState",
	/* LWTRANCHE_REPLICATION_SLOT_IO: */
	"ReplicationSlotIO",
	/* LWTRANCHE_LOCK_FASTPATH: */
	"LockFastPath",
	/* LWTRANCHE_BUFFER_MAPPING: */
	"BufferMapping",
	/* LWTRANCHE_LOCK_MANAGER: */
	"LockManager",
	/* LWTRANCHE_PREDICATE_LOCK_MANAGER: */
	"PredicateLockManager",
	/* LWTRANCHE_PARALLEL_HASH_JOIN: */
	"ParallelHashJoin",
	/* LWTRANCHE_PARALLEL_QUERY_DSA: */
	"ParallelQueryDSA",
	/* LWTRANCHE_PER_SESSION_DSA: */
	"PerSessionDSA",
	/* LWTRANCHE_PER_SESSION_RECORD_TYPE: */
	"PerSessionRecordType",
	/* LWTRANCHE_PER_SESSION_RECORD_TYPMOD: */
	"PerSessionRecordTypmod",
	/* LWTRANCHE_SHARED_TUPLESTORE: */
	"SharedTupleStore",
	/* LWTRANCHE_SHARED_TIDBITMAP: */
	"SharedTidBitmap",
	/* LWTRANCHE_PARALLEL_APPEND: */
	"ParallelAppend",
	/* LWTRANCHE_PER_XACT_PREDICATE_LIST: */
	"PerXactPredicateList",
	/* LWTRANCHE_PGSTATS_DSA: */
	"PgStatsDSA",
	/* LWTRANCHE_PGSTATS_HASH: */
	"PgStatsHash",
	/* LWTRANCHE_PGSTATS_DATA: */
	"PgStatsData",
};

StaticAssertDecl(lengthof(BuiltinTrancheNames) ==
				 LWTRANCHE_FIRST_USER_DEFINED - NUM_INDIVIDUAL_LWLOCKS,
				 "missing entries in BuiltinTrancheNames[]");

/*
 * 这是通过减去 LWTRANCHE_FIRST_USER_DEFINED 的 tranche ID 索引的， 
 * 存储当前进程已知的所有动态创建的 tranches 的名称。 
 * 数组中任何未使用的条目将包含 NULL。
 */
static const char **LWLockTrancheNames = NULL;
static int	LWLockTrancheNamesAllocated = 0;

/*
 * 这指向共享内存中 LWLocks 的主数组。 后端通过从 postmaster 进行 fork 继承
 * 指针（在 EXEC_BACKEND 情况下除外，我们有特殊措施将其传递下去）。
 */
LWLockPadded *MainLWLockArray = NULL;

/*
 * 我们使用这个结构来跟踪在错误恢复期间释放的锁定 LWLocks。
 * 通常一次只保持几个，但有时数量可以高得多；例如，pg_buffercache
 * 扩展同时锁定所有缓存分区。
 */
#define MAX_SIMUL_LWLOCKS	200

/* 表示我们持有的 LWLocks 的结构 */
typedef struct LWLockHandle
{
	LWLock	   *lock;
	LWLockMode	mode;
} LWLockHandle;

static int	num_held_lwlocks = 0;
static LWLockHandle held_lwlocks[MAX_SIMUL_LWLOCKS];

/* 表示为命名 tranche 请求的 LWLock tranche 的结构 */
typedef struct NamedLWLockTrancheRequest
{
	char		tranche_name[NAMEDATALEN];
	int			num_lwlocks;
} NamedLWLockTrancheRequest;

static NamedLWLockTrancheRequest *NamedLWLockTrancheRequestArray = NULL;
static int	NamedLWLockTrancheRequestsAllocated = 0;

/*
 * NamedLWLockTrancheRequests 是请求数组的有效长度，
 * 也是稍后共享内存 NamedLWLockTrancheArray 的长度。
 * 这个变量和 NamedLWLockTrancheArray 是非静态的，这样 
 * postmaster.c 可以在 EXEC_BACKEND 构建中将它们复制到子进程。
 */
int			NamedLWLockTrancheRequests = 0;

/* 指向共享内存中的数据: */
NamedLWLockTranche *NamedLWLockTrancheArray = NULL;

static void fc_InitializeLWLocks(void);
static inline void fc_LWLockReportWaitStart(LWLock *fc_lock);
static inline void fc_LWLockReportWaitEnd(void);
static const char *fc_GetLWTrancheName(uint16 fc_trancheId);

#define T_NAME(lock) \
	fc_GetLWTrancheName((lock)->tranche)

#ifdef LWLOCK_STATS
typedef struct lwlock_stats_key
{
	int			tranche;
	void	   *instance;
}			lwlock_stats_key;

typedef struct lwlock_stats
{
	lwlock_stats_key key;
	int			sh_acquire_count;
	int			ex_acquire_count;
	int			block_count;
	int			dequeue_self_count;
	int			spin_delay_count;
}			lwlock_stats;

static HTAB *lwlock_stats_htab;
static lwlock_stats lwlock_stats_dummy;
#endif

#ifdef LOCK_DEBUG
bool		Trace_lwlocks = false;

inline static void fc_PRINT_LWDEBUG(const char *fc_where, LWLock *fc_lock, LWLockMode fc_mode)
{
	/* 在这里隐藏语句和上下文，否则日志会太冗长 */
	if (Trace_lwlocks)
	{
		uint32		fc_state = pg_atomic_read_u32(&fc_lock->state);

		ereport(LOG,
				(errhidestmt(true),
				 errhidecontext(true),
				 errmsg_internal("%d: %s(%s %p): excl %u shared %u haswaiters %u waiters %u rOK %d",
								 MyProcPid,
								 fc_where, T_NAME(fc_lock), fc_lock,
								 (fc_state & LW_VAL_EXCLUSIVE) != 0,
								 fc_state & LW_SHARED_MASK,
								 (fc_state & LW_FLAG_HAS_WAITERS) != 0,
								 pg_atomic_read_u32(&fc_lock->nwaiters),
								 (fc_state & LW_FLAG_RELEASE_OK) != 0)));
	}
}

inline static void fc_LOG_LWDEBUG(const char *fc_where, LWLock *fc_lock, const char *fc_msg)
{
	/* 在这里隐藏语句和上下文，否则日志会太冗长 */
	if (Trace_lwlocks)
	{
		ereport(LOG,
				(errhidestmt(true),
				 errhidecontext(true),
				 errmsg_internal("%s(%s %p): %s", fc_where,
								 T_NAME(fc_lock), fc_lock, fc_msg)));
	}
}

#else							/* 不是 LOCK_DEBUG */
#define fc_PRINT_LWDEBUG(a,b,c) ((void)0)
#define fc_LOG_LWDEBUG(a,b,c) ((void)0)
#endif							/* LOCK_DEBUG */

#ifdef LWLOCK_STATS

static void fc_init_lwlock_stats(void);
static void fc_print_lwlock_stats(int fc_code, Datum fc_arg);
static lwlock_stats * get_lwlock_stats_entry(LWLock *fc_lock);

static void fc_init_lwlock_stats(void)
{
	HASHCTL		fc_ctl;
	static MemoryContext fc_lwlock_stats_cxt = NULL;
	static bool fc_exit_registered = false;

	if (fc_lwlock_stats_cxt != NULL)
		MemoryContextDelete(fc_lwlock_stats_cxt);

	/*
	 * LWLock 统计信息将在关键区段中更新，这需要分配新的哈希条目。
	 * 在关键区段内通常不允许进行分配，因为内存不足将导致 PANIC，
	 * 但是 LWLOCK_STATS 是调试代码，通常在生产中不会打开，
	 * 因此这是一个可以接受的风险。哈希条目很小，
	 * 因此在实践中运行内存不足的风险是最小的。
	 */
	fc_lwlock_stats_cxt = AllocSetContextCreate(TopMemoryContext,
											 "LWLock stats",
											 ALLOCSET_DEFAULT_SIZES);
	MemoryContextAllowInCriticalSection(fc_lwlock_stats_cxt, true);

	fc_ctl.keysize = sizeof(lwlock_stats_key);
	fc_ctl.entrysize = sizeof(lwlock_stats);
	fc_ctl.hcxt = fc_lwlock_stats_cxt;
	lwlock_stats_htab = hash_create("lwlock stats", 16384, &fc_ctl,
									HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);
	if (!fc_exit_registered)
	{
		on_shmem_exit(fc_print_lwlock_stats, 0);
		fc_exit_registered = true;
	}
}

static void fc_print_lwlock_stats(int fc_code, Datum fc_arg)
{
	HASH_SEQ_STATUS fc_scan;
	lwlock_stats *fc_lwstats;

	hash_seq_init(&fc_scan, lwlock_stats_htab);

	/* 获取一个 LWLock，以防止不同后端混合报告 */
	LWLockAcquire(&MainLWLockArray[0].lock, LW_EXCLUSIVE);

	while ((fc_lwstats = (lwlock_stats *) hash_seq_search(&fc_scan)) != NULL)
	{
		fprintf(stderr,
#ifdef FDD //cppcheck
				"PID %d lwlock %s %p: shacq %d exacq %d blk %d spindelay %d dequeue self %d\n",
#else
				"PID %d lwlock %s %p: shacq %u exacq %u blk %u spindelay %u dequeue self %u\n",
#endif
				MyProcPid, fc_GetLWTrancheName(fc_lwstats->key.tranche),
				fc_lwstats->key.instance, fc_lwstats->sh_acquire_count,
				fc_lwstats->ex_acquire_count, fc_lwstats->block_count,
				fc_lwstats->spin_delay_count, fc_lwstats->dequeue_self_count);
	}

	LWLockRelease(&MainLWLockArray[0].lock);
}

static lwlock_stats *
get_lwlock_stats_entry(LWLock *fc_lock)
{
	lwlock_stats_key fc_key;
	lwlock_stats *fc_lwstats;
	bool		fc_found;

	/*
	 * 在共享内存初始化期间，哈希表尚不存在。
	 * 那个阶段的统计并不太有趣，所以只需在一个单一的虚拟条目中收集所有锁的操作。
	 */
	if (lwlock_stats_htab == NULL)
		return &lwlock_stats_dummy;

	/* 获取或创建条目。 */
	MemSet(&fc_key, 0, sizeof(fc_key));
	fc_key.tranche = fc_lock->tranche;
	fc_key.instance = fc_lock;
	fc_lwstats = hash_search(lwlock_stats_htab, &fc_key, HASH_ENTER, &fc_found);
	if (!fc_found)
	{
		fc_lwstats->sh_acquire_count = 0;
		fc_lwstats->ex_acquire_count = 0;
		fc_lwstats->block_count = 0;
		fc_lwstats->dequeue_self_count = 0;
		fc_lwstats->spin_delay_count = 0;
	}
	return fc_lwstats;
}
#endif							/* LWLOCK_STATS */


/*
 * 计算命名 tranches 所需的 LWLocks 数量。 这些将在主数组中分配。
 */
static int fc_NumLWLocksForNamedTranches(void)
{
	int			fc_numLocks = 0;
	int			fc_i;

	for (fc_i = 0; fc_i < NamedLWLockTrancheRequests; fc_i++)
		fc_numLocks += NamedLWLockTrancheRequestArray[fc_i].num_lwlocks;

	return fc_numLocks;
}

/*
 * 计算 LWLocks 和命名 tranches 所需的共享内存空间。
 */
Size LWLockShmemSize(void)
{
	Size		fc_size;
	int			fc_i;
	int			fc_numLocks = NUM_FIXED_LWLOCKS;

	/* 计算主数组中需要的锁的总数。 */
	fc_numLocks += fc_NumLWLocksForNamedTranches();

	/* LWLock 数组的空间。 */
	fc_size = mul_size(fc_numLocks, sizeof(LWLockPadded));

	/* 动态分配计数器的空间，加上对齐的空间。 */
	fc_size = add_size(fc_size, sizeof(int) + LWLOCK_PADDED_SIZE);

	/* 名称 tranches 的空间。 */
	fc_size = add_size(fc_size, mul_size(NamedLWLockTrancheRequests, sizeof(NamedLWLockTranche)));

	/* 每个 tranche 名称的空间。 */
	for (fc_i = 0; fc_i < NamedLWLockTrancheRequests; fc_i++)
		fc_size = add_size(fc_size, strlen(NamedLWLockTrancheRequestArray[fc_i].tranche_name) + 1);

	return fc_size;
}

/*
 * 分配共享内存空间以用于主 LWLock 数组和所有 tranches 并
 * 初始化它。 我们还在这里注册扩展 LWLock tranches。
 */
void CreateLWLocks(void)
{
	StaticAssertStmt(LW_VAL_EXCLUSIVE > (uint32) MAX_BACKENDS,
					 "MAX_BACKENDS too big for lwlock.c");

	StaticAssertStmt(sizeof(LWLock) <= LWLOCK_PADDED_SIZE,
					 "Miscalculated LWLock padding");

	if (!IsUnderPostmaster)
	{
		Size		fc_spaceLocks = LWLockShmemSize();
		int		   *fc_LWLockCounter;
		char	   *fc_ptr;

		/* 分配空间 */
		fc_ptr = (char *) ShmemAlloc(fc_spaceLocks);

		/* 留出动态分配 tranches 的空间 */
		fc_ptr += sizeof(int);

		/* 确保 LWLock 数组的所需对齐 */
		fc_ptr += LWLOCK_PADDED_SIZE - ((uintptr_t) fc_ptr) % LWLOCK_PADDED_SIZE;

		MainLWLockArray = (LWLockPadded *) fc_ptr;

		/*
		 * 初始化 tranche 的动态分配计数器，该计数器
		 * 存储在第一个 LWLock 之前。
		 */
		fc_LWLockCounter = (int *) ((char *) MainLWLockArray - sizeof(int));
		*fc_LWLockCounter = LWTRANCHE_FIRST_USER_DEFINED;

		/* 初始化所有 LWLocks */
		fc_InitializeLWLocks();
	}

	/* 在当前进程中注册命名扩展 LWLock tranches。 */
	for (int fc_i = 0; fc_i < NamedLWLockTrancheRequests; fc_i++)
		LWLockRegisterTranche(NamedLWLockTrancheArray[fc_i].trancheId,
							  NamedLWLockTrancheArray[fc_i].trancheName);
}

/*
 * 初始化固定的 LWLocks 和属于命名区段的 LWLocks。
 */
static void fc_InitializeLWLocks(void)
{
	int			fc_numNamedLocks = fc_NumLWLocksForNamedTranches();
	int			fc_id;
	int			fc_i;
	int			fc_j;
	LWLockPadded *fc_lock;

	/* 初始化主数组中的所有单个 LWLocks */
	for (fc_id = 0, fc_lock = MainLWLockArray; fc_id < NUM_INDIVIDUAL_LWLOCKS; fc_id++, fc_lock++)
		LWLockInitialize(&fc_lock->lock, fc_id);

	/* 初始化主数组中的缓冲区映射 LWLocks */
	fc_lock = MainLWLockArray + BUFFER_MAPPING_LWLOCK_OFFSET;
	for (fc_id = 0; fc_id < NUM_BUFFER_PARTITIONS; fc_id++, fc_lock++)
		LWLockInitialize(&fc_lock->lock, LWTRANCHE_BUFFER_MAPPING);

	/* 初始化主数组中的 lmgrs 的 LWLocks */
	fc_lock = MainLWLockArray + LOCK_MANAGER_LWLOCK_OFFSET;
	for (fc_id = 0; fc_id < NUM_LOCK_PARTITIONS; fc_id++, fc_lock++)
		LWLockInitialize(&fc_lock->lock, LWTRANCHE_LOCK_MANAGER);

	/* 初始化主数组中的谓词 lmgrs 的 LWLocks */
	fc_lock = MainLWLockArray + PREDICATELOCK_MANAGER_LWLOCK_OFFSET;
	for (fc_id = 0; fc_id < NUM_PREDICATELOCK_PARTITIONS; fc_id++, fc_lock++)
		LWLockInitialize(&fc_lock->lock, LWTRANCHE_PREDICATE_LOCK_MANAGER);

	/*
	 * 将关于任何命名区段的信息复制到共享内存中（以便
	 * 其他进程可以看到它），并初始化请求的 LWLocks。
	 */
	if (NamedLWLockTrancheRequests > 0)
	{
		char	   *fc_trancheNames;

		NamedLWLockTrancheArray = (NamedLWLockTranche *)
			&MainLWLockArray[NUM_FIXED_LWLOCKS + fc_numNamedLocks];

		fc_trancheNames = (char *) NamedLWLockTrancheArray +
			(NamedLWLockTrancheRequests * sizeof(NamedLWLockTranche));
		fc_lock = &MainLWLockArray[NUM_FIXED_LWLOCKS];

		for (fc_i = 0; fc_i < NamedLWLockTrancheRequests; fc_i++)
		{
			NamedLWLockTrancheRequest *fc_request;
			NamedLWLockTranche *fc_tranche;
			char	   *fc_name;

			fc_request = &NamedLWLockTrancheRequestArray[fc_i];
			fc_tranche = &NamedLWLockTrancheArray[fc_i];

			fc_name = fc_trancheNames;
			fc_trancheNames += strlen(fc_request->tranche_name) + 1;
			strcpy(fc_name, fc_request->tranche_name);
			fc_tranche->trancheId = LWLockNewTrancheId();
			fc_tranche->trancheName = fc_name;

			for (fc_j = 0; fc_j < fc_request->num_lwlocks; fc_j++, fc_lock++)
				LWLockInitialize(&fc_lock->lock, fc_tranche->trancheId);
		}
	}
}

/*
 * InitLWLockAccess - 初始化持有 LWLocks 所需的后端本地状态
 */
void InitLWLockAccess(void)
{
#ifdef LWLOCK_STATS
	fc_init_lwlock_stats();
#endif
}

/*
 * GetNamedLWLockTranche - 返回指定区段中 LWLock 的基地址。
 *
 * 调用者需要从此 API 返回的基锁地址开始检索请求的 LWLocks 数量。
 * 这可以用于通过使用 RequestNamedLWLockTranche() API 请求的区段。
 */
LWLockPadded *
GetNamedLWLockTranche(const char *fc_tranche_name)
{
	int			fc_lock_pos;
	int			fc_i;

	/*
	 * 获取主 LWLock 数组中属于请求的
	 * tranche_name 的 LWLock 的基地址的位置。命名区段的 LWLocks
	 * 被放置在固定锁之后的主 LWLock 数组中。
	 */
	fc_lock_pos = NUM_FIXED_LWLOCKS;
	for (fc_i = 0; fc_i < NamedLWLockTrancheRequests; fc_i++)
	{
		if (strcmp(NamedLWLockTrancheRequestArray[fc_i].tranche_name,
				   fc_tranche_name) == 0)
			return &MainLWLockArray[fc_lock_pos];

		fc_lock_pos += NamedLWLockTrancheRequestArray[fc_i].num_lwlocks;
	}

	elog(ERROR, "requested tranche is not registered");

	/* 只是为了让编译器安静 */
	return NULL;
}

/*
 * 分配一个新的区段 ID。
 */
int LWLockNewTrancheId(void)
{
	int			fc_result;
	int		   *fc_LWLockCounter;

	fc_LWLockCounter = (int *) ((char *) MainLWLockArray - sizeof(int));
	SpinLockAcquire(ShmemLock);
	fc_result = (*fc_LWLockCounter)++;
	SpinLockRelease(ShmemLock);

	return fc_result;
}

/*
 * 在当前进程的查找表中注册一个动态区段名称。
 *
 * 此例程将保存作为参数传递的区段名称的指针，
 * 因此名称应在后端生命周期上下文中分配
 * （共享内存、TopMemoryContext、静态常量或类似项目）。
 *
 * 区段名称将在用户可见中作为等待事件名称，因此尽量使用
 * 符合样式的名称。
 */
void LWLockRegisterTranche(int fc_tranche_id, const char *fc_tranche_name)
{
	/* 这应该仅在用户定义的区段中调用。 */
	if (fc_tranche_id < LWTRANCHE_FIRST_USER_DEFINED)
		return;

	/* 转换为数组索引。 */
	fc_tranche_id -= LWTRANCHE_FIRST_USER_DEFINED;

	/* 如有必要，创建或扩大数组。 */
	if (fc_tranche_id >= LWLockTrancheNamesAllocated)
	{
		int			fc_newalloc;

		fc_newalloc = pg_nextpower2_32(Max(8, fc_tranche_id + 1));

		if (LWLockTrancheNames == NULL)
			LWLockTrancheNames = (const char **)
				MemoryContextAllocZero(TopMemoryContext,
									   fc_newalloc * sizeof(char *));
		else
		{
			LWLockTrancheNames = (const char **)
				repalloc(LWLockTrancheNames, fc_newalloc * sizeof(char *));
			memset(LWLockTrancheNames + LWLockTrancheNamesAllocated,
				   0,
				   (fc_newalloc - LWLockTrancheNamesAllocated) * sizeof(char *));
		}
		LWLockTrancheNamesAllocated = fc_newalloc;
	}

	LWLockTrancheNames[fc_tranche_id] = fc_tranche_name;
}

/*
 * RequestNamedLWLockTranche
 *		请求在后端启动期间分配额外的 LWLocks。
 *
 * 这只能通过通过 shared_preload_libraries 加载到后端的库的 shmem_request_hook 来调用。
 * 从其他地方的调用将失败。
 *
 * 区段名称将在用户可见中作为等待事件名称，因此尽量使用
 * 符合样式的名称。
 */
void RequestNamedLWLockTranche(const char *fc_tranche_name, int fc_num_lwlocks)
{
	NamedLWLockTrancheRequest *fc_request;

	if (!process_shmem_requests_in_progress)
		elog(FATAL, "cannot request additional LWLocks outside shmem_request_hook");

	if (NamedLWLockTrancheRequestArray == NULL)
	{
		NamedLWLockTrancheRequestsAllocated = 16;
		NamedLWLockTrancheRequestArray = (NamedLWLockTrancheRequest *)
			MemoryContextAlloc(TopMemoryContext,
							   NamedLWLockTrancheRequestsAllocated
							   * sizeof(NamedLWLockTrancheRequest));
	}

	if (NamedLWLockTrancheRequests >= NamedLWLockTrancheRequestsAllocated)
	{
		int			fc_i = pg_nextpower2_32(NamedLWLockTrancheRequests + 1);

		NamedLWLockTrancheRequestArray = (NamedLWLockTrancheRequest *)
			repalloc(NamedLWLockTrancheRequestArray,
					 fc_i * sizeof(NamedLWLockTrancheRequest));
		NamedLWLockTrancheRequestsAllocated = fc_i;
	}

	fc_request = &NamedLWLockTrancheRequestArray[NamedLWLockTrancheRequests];
	Assert(strlen(fc_tranche_name) + 1 <= NAMEDATALEN);
	strlcpy(fc_request->tranche_name, fc_tranche_name, NAMEDATALEN);
	fc_request->num_lwlocks = fc_num_lwlocks;
	NamedLWLockTrancheRequests++;
}

/*
 * LWLockInitialize - 初始化一个新的 lwlock；它最初是解锁的
 */
void LWLockInitialize(LWLock *fc_lock, int fc_tranche_id)
{
	pg_atomic_init_u32(&fc_lock->state, LW_FLAG_RELEASE_OK);
#ifdef LOCK_DEBUG
	pg_atomic_init_u32(&fc_lock->nwaiters, 0);
#endif
	fc_lock->tranche = fc_tranche_id;
	proclist_init(&fc_lock->waiters);
}

/*
 * 报告轻量级锁的等待事件开始。
 *
 * 此函数将被所有需要等待以获取锁的轻量级锁调用使用。此函数根据区段和锁 ID 区分等待事件。
 */
static inline void fc_LWLockReportWaitStart(LWLock *fc_lock)
{
	pgstat_report_wait_start(PG_WAIT_LWLOCK | fc_lock->tranche);
}

/*
 * 报告轻量级锁的等待事件结束。
 */
static inline void fc_LWLockReportWaitEnd(void)
{
	pgstat_report_wait_end();
}

/*
 * 返回 LWLock 区段的名称。
 */
static const char * fc_GetLWTrancheName(uint16 fc_trancheId)
{
	/* 单个 LWLock？ */
	if (fc_trancheId < NUM_INDIVIDUAL_LWLOCKS)
		return IndividualLWLockNames[fc_trancheId];

	/* 内置区段？ */
	if (fc_trancheId < LWTRANCHE_FIRST_USER_DEFINED)
		return BuiltinTrancheNames[fc_trancheId - NUM_INDIVIDUAL_LWLOCKS];

	/*
	 * 这是一个扩展区段，因此在 LWLockTrancheNames[] 中查找。然而，
	 * 可能该区段在当前进程中从未注册，在这种情况下放弃并返回 "extension"。
	 */
	fc_trancheId -= LWTRANCHE_FIRST_USER_DEFINED;

	if (fc_trancheId >= LWLockTrancheNamesAllocated ||
		LWLockTrancheNames[fc_trancheId] == NULL)
		return "extension";

	return LWLockTrancheNames[fc_trancheId];
}

/*
 * 根据等待类和事件返回 LWLock 的标识符。
 */
const char * GetLWLockIdentifier(uint32 fc_classId, uint16 fc_eventId)
{
	Assert(fc_classId == PG_WAIT_LWLOCK);
	/* 事件 ID 只是区段编号。 */
	return fc_GetLWTrancheName(fc_eventId);
}

/*
 * 内部函数，尝试在传入模式下原子地获取 lwlock。
 *
 * 此函数不会阻止等待锁变得可用 - 这是调用者的工作。
 *
 * 如果锁未释放且我们需要等待，则返回 true。
 */
static bool fc_LWLockAttemptLock(LWLock *fc_lock, LWLockMode fc_mode)
{
	uint32		fc_old_state;

	AssertArg(fc_mode == LW_EXCLUSIVE || fc_mode == LW_SHARED);

	/*
	 * 在循环外部读取一次，后面的迭代将通过比较和交换获取更新值。
	 */
	fc_old_state = pg_atomic_read_u32(&fc_lock->state);

	/* 循环直到我们确定是否能够获取锁 */
	while (true)
	{
		uint32		fc_desired_state;
		bool		fc_lock_free;

		fc_desired_state = fc_old_state;

		if (fc_mode == LW_EXCLUSIVE)
		{
			fc_lock_free = (fc_old_state & LW_LOCK_MASK) == 0;
			if (fc_lock_free)
				fc_desired_state += LW_VAL_EXCLUSIVE;
		}
		else
		{
			fc_lock_free = (fc_old_state & LW_VAL_EXCLUSIVE) == 0;
			if (fc_lock_free)
				fc_desired_state += LW_VAL_SHARED;
		}

		/*
		 * 尝试交换我们期望的状态。如果我们没有看到锁是自由的，那只是旧值。如果我们看到它是自由的，
		 * 我们将尝试标记为已获取。我们始终交换值的原因在于这也充当内存屏障。我们可以尝试
		 * 更聪明，只在我们看到锁是自由时交换值，但基准测试到目前为止尚未显示这有益。
		 *
		 * 如果自上次查看以来值已更改，则重试。
		 */
		if (pg_atomic_compare_exchange_u32(&fc_lock->state,
										   &fc_old_state, fc_desired_state))
		{
			if (fc_lock_free)
			{
				/* 太好了！得到了锁。 */
#ifdef LOCK_DEBUG
				if (fc_mode == LW_EXCLUSIVE)
					fc_lock->owner = MyProc;
#endif
				return false;
			}
			else
				return true;	/* 其他人已获取锁 */
		}
	}
	pg_unreachable();
}

/*
 * 锁定 LWLock 的等待列表以防止并发活动。
 *
 * 注意：即使等待列表被锁定，非冲突的锁操作仍然可以并发发生。
 *
 * 持有互斥锁的时间应当尽量短暂！
 */
static void fc_LWLockWaitListLock(LWLock *fc_lock)
{
	uint32		fc_old_state;
#ifdef LWLOCK_STATS
	lwlock_stats *fc_lwstats;
	uint32		fc_delays = 0;

	fc_lwstats = get_lwlock_stats_entry(fc_lock);
#endif

	while (true)
	{
		/* 始终尝试一次直接获取锁 */
		fc_old_state = pg_atomic_fetch_or_u32(&fc_lock->state, LW_FLAG_LOCKED);
		if (!(fc_old_state & LW_FLAG_LOCKED))
			break;				/* got lock */

		/* 然后无原子操作地旋转，直到锁被释放 */
		{
			SpinDelayStatus fc_delayStatus;

			init_local_spin_delay(&fc_delayStatus);

			while (fc_old_state & LW_FLAG_LOCKED)
			{
				perform_spin_delay(&fc_delayStatus);
				fc_old_state = pg_atomic_read_u32(&fc_lock->state);
			}
#ifdef LWLOCK_STATS
			fc_delays += fc_delayStatus.delays;
#endif
			finish_spin_delay(&fc_delayStatus);
		}

		/*
		 * 重试。到我们再次尝试获取锁时，锁可能已经被重新获取。
		 */
	}

#ifdef LWLOCK_STATS
	fc_lwstats->spin_delay_count += fc_delays;
#endif
}

/*
 * 解锁 LWLock 的等待列表。
 *
 * 注意，可以通过单个原子操作来操纵标志并释放锁，这样可能会更高效。
 */
static void fc_LWLockWaitListUnlock(LWLock *fc_lock)
{
	uint32		fc_old_state PG_USED_FOR_ASSERTS_ONLY;

	fc_old_state = pg_atomic_fetch_and_u32(&fc_lock->state, ~LW_FLAG_LOCKED);

	Assert(fc_old_state & LW_FLAG_LOCKED);
}

/*
 * 唤醒所有当前有机会获取锁的锁持有者。
 */
static void fc_LWLockWakeup(LWLock *fc_lock)
{
	bool		fc_new_release_ok;
	bool		fc_wokeup_somebody = false;
	proclist_head fc_wakeup;
	proclist_mutable_iter fc_iter;

	proclist_init(&fc_wakeup);

	fc_new_release_ok = true;

	/* 在收集需要被唤醒的后端时锁定等待列表 */
	fc_LWLockWaitListLock(fc_lock);

	proclist_foreach_modify(fc_iter, &fc_lock->waiters, lwWaitLink)
	{
		PGPROC	   *fc_waiter = GetPGProcByNumber(fc_iter.cur);

		if (fc_wokeup_somebody && fc_waiter->lwWaitMode == LW_EXCLUSIVE)
			continue;

		proclist_delete(&fc_lock->waiters, fc_iter.cur, lwWaitLink);
		proclist_push_tail(&fc_wakeup, fc_iter.cur, lwWaitLink);

		if (fc_waiter->lwWaitMode != LW_WAIT_UNTIL_FREE)
		{
			/*
			 * 在重试者运行之前防止额外的唤醒。仅仅在等待锁变为空闲的后端
			 * 不会自动重试。
			 */
			fc_new_release_ok = false;

			/*
			 * 不要再唤醒（进一步的）独占锁。
			 */
			fc_wokeup_somebody = true;
		}

		/*
		 * 发信号表示该进程不再在等待列表中。这使得
		 * LWLockDequeueSelf() 可以通过 proclist_delete() 
		 * 将自己从等待列表中移除，而不必检查自己是否已经
		 * 从列表中移除了。
		 */
		Assert(fc_waiter->lwWaiting == LW_WS_WAITING);
		fc_waiter->lwWaiting = LW_WS_PENDING_WAKEUP;

		/*
		 * 一旦我们唤醒了一个独占锁，唤醒其他用户就没有意义了。
		 */
		if (fc_waiter->lwWaitMode == LW_EXCLUSIVE)
			break;
	}

	Assert(proclist_is_empty(&fc_wakeup) || pg_atomic_read_u32(&fc_lock->state) & LW_FLAG_HAS_WAITERS);

	/* 一次性取消所需标志并释放锁 */
	{
		uint32		fc_old_state;
		uint32		fc_desired_state;

		fc_old_state = pg_atomic_read_u32(&fc_lock->state);
		while (true)
		{
			fc_desired_state = fc_old_state;

			/* 计算期望的标志 */

			if (fc_new_release_ok)
				fc_desired_state |= LW_FLAG_RELEASE_OK;
			else
				fc_desired_state &= ~LW_FLAG_RELEASE_OK;

			if (proclist_is_empty(&fc_wakeup))
				fc_desired_state &= ~LW_FLAG_HAS_WAITERS;

			fc_desired_state &= ~LW_FLAG_LOCKED;	/* 释放锁 */

			if (pg_atomic_compare_exchange_u32(&fc_lock->state, &fc_old_state,
											   fc_desired_state))
				break;
		}
	}

	/* 唤醒我从队列中移除的任何等待者。 */
	proclist_foreach_modify(fc_iter, &fc_wakeup, lwWaitLink)
	{
		PGPROC	   *fc_waiter = GetPGProcByNumber(fc_iter.cur);

		fc_LOG_LWDEBUG("LWLockRelease", fc_lock, "release waiter");
		proclist_delete(&fc_wakeup, fc_iter.cur, lwWaitLink);

		/*
		 * 确保 lwWaiting 被取消的可见性仅在
		 * 从链中断开后才能实现。否则，目标后端
		 * 可能会因其他原因被唤醒，并排队等待新锁 - 如果
		 * 这在列表断开之前发生，列表最终会变得
		 * 损坏。
		 *
		 * 该屏障在为另一个锁排队时与 LWLockWaitListLock() 配对。
		 */
		pg_write_barrier();
		fc_waiter->lwWaiting = LW_WS_NOT_WAITING;
		PGSemaphoreUnlock(fc_waiter->sem);
	}
}

/*
 * 将自己添加到队列的末尾。
 *
 * 注意：模式可以是 LW_WAIT_UNTIL_FREE！
 */
static void fc_LWLockQueueSelf(LWLock *fc_lock, LWLockMode fc_mode)
{
	/*
	 * 如果我们没有 PGPROC 结构，就没有办法等待。这
	 * 不应发生，因为 MyProc 仅在共享
	 * 内存初始化期间为 null。
	 */
	if (MyProc == NULL)
		elog(PANIC, "cannot wait without a PGPROC structure");

	if (MyProc->lwWaiting != LW_WS_NOT_WAITING)
		elog(PANIC, "queueing for lock while waiting on another one");

	fc_LWLockWaitListLock(fc_lock);

	/* 设置标志受自旋锁保护 */
	pg_atomic_fetch_or_u32(&fc_lock->state, LW_FLAG_HAS_WAITERS);

	MyProc->lwWaiting = LW_WS_WAITING;
	MyProc->lwWaitMode = fc_mode;

	/* LW_WAIT_UNTIL_FREE 等待者始终在队列的前面 */
	if (fc_mode == LW_WAIT_UNTIL_FREE)
		proclist_push_head(&fc_lock->waiters, MyProc->pgprocno, lwWaitLink);
	else
		proclist_push_tail(&fc_lock->waiters, MyProc->pgprocno, lwWaitLink);

	/* 现在可以释放互斥锁 */
	fc_LWLockWaitListUnlock(fc_lock);

#ifdef LOCK_DEBUG
	pg_atomic_fetch_add_u32(&fc_lock->nwaiters, 1);
#endif
}

/*
 * 从等待列表中删除自己。
 *
 * 如果我们是因为认为需要睡眠而将自己排队，但在进一步检查之后我们发现实际上不需要这样
 * ，则使用此语句。
 */
static void fc_LWLockDequeueSelf(LWLock *fc_lock)
{
	bool		fc_on_waitlist;

#ifdef LWLOCK_STATS
	lwlock_stats *fc_lwstats;

	fc_lwstats = get_lwlock_stats_entry(fc_lock);

	fc_lwstats->dequeue_self_count++;
#endif

	fc_LWLockWaitListLock(fc_lock);

	/*
	 * 除非我们已经被移除，否则将自己从等待列表中移除。
	 * 移除是在持有等待列表锁的情况下发生的，因此在此检查中没有竞争。
	 */
	fc_on_waitlist = MyProc->lwWaiting == LW_WS_WAITING;
	if (fc_on_waitlist)
		proclist_delete(&fc_lock->waiters, MyProc->pgprocno, lwWaitLink);

	if (proclist_is_empty(&fc_lock->waiters) &&
		(pg_atomic_read_u32(&fc_lock->state) & LW_FLAG_HAS_WAITERS) != 0)
	{
		pg_atomic_fetch_and_u32(&fc_lock->state, ~LW_FLAG_HAS_WAITERS);
	}

	/* XXX: 与上面的 fetch_and 合并？ */
	fc_LWLockWaitListUnlock(fc_lock);

	/* 再次清除等待状态，便于调试 */
	if (fc_on_waitlist)
		MyProc->lwWaiting = LW_WS_NOT_WAITING;
	else
	{
		int			fc_extraWaits = 0;

		/*
		 * 其他人将我们从队列中移除并将唤醒我们。处理
		 * 无谓的唤醒吸收。
		 */

		/*
		 * 如果在我们移除自己之前有人唤醒了我们，则重置 RELEASE_OK 标志 - 
		 * 他们会将其设置为 false。
		 */
		pg_atomic_fetch_or_u32(&fc_lock->state, LW_FLAG_RELEASE_OK);

		/*
		 * 现在等待预定的唤醒，否则我们的 ->lwWaiting 会在后面的某个不方便的点
		 * 被重置。大多数情况下，这将立即返回。
		 */
		for (;;)
		{
			PGSemaphoreLock(MyProc->sem);
			if (MyProc->lwWaiting == LW_WS_NOT_WAITING)
				break;
			fc_extraWaits++;
		}

		/*
		 * 修正进程等待信号量的计数以吸收任何唤醒。
		 */
		while (fc_extraWaits-- > 0)
			PGSemaphoreUnlock(MyProc->sem);
	}

#ifdef LOCK_DEBUG
	{
		/* 不再等待 */
		uint32		fc_nwaiters PG_USED_FOR_ASSERTS_ONLY = pg_atomic_fetch_sub_u32(&fc_lock->nwaiters, 1);

		Assert(fc_nwaiters < MAX_BACKENDS);
	}
#endif
}

/*
 * LWLockAcquire - 以指定模式获取轻量级锁
 *
 * 如果锁不可用，则等待直到它可用。如果锁立即可用则返回true，如果我们不得不等待则返回false。
 *
 * 侧面影响：取消/死亡中断在锁释放之前被延迟。
 */
bool LWLockAcquire(LWLock *fc_lock, LWLockMode fc_mode)
{
	PGPROC	   *fc_proc = MyProc;
	bool		fc_result = true;
	int			fc_extraWaits = 0;
#ifdef LWLOCK_STATS
	lwlock_stats *fc_lwstats;

	fc_lwstats = get_lwlock_stats_entry(fc_lock);
#endif

	AssertArg(fc_mode == LW_SHARED || fc_mode == LW_EXCLUSIVE);

	fc_PRINT_LWDEBUG("LWLockAcquire", fc_lock, fc_mode);

#ifdef LWLOCK_STATS
	/* 统计锁获取尝试 */
	if (fc_mode == LW_EXCLUSIVE)
		fc_lwstats->ex_acquire_count++;
	else
		fc_lwstats->sh_acquire_count++;
#endif							/* LWLOCK_STATS */

	/*
	 * 如果我们没有获得PGPROC，则不能等待。这应该只发生在引导或共享内存初始化期间。在这里放置一个Assert以捕捉不安全的编码实践。
	 */
	Assert(!(fc_proc == NULL && IsUnderPostmaster));

	/* 确保我们将有空间记住锁 */
	if (num_held_lwlocks >= MAX_SIMUL_LWLOCKS)
		elog(ERROR, "too many LWLocks taken");

	/*
	 * 在退出受LWLock保护的代码部分之前锁定取消/死亡中断。这确保中断不会干扰对共享内存中数据结构的操作。
	 */
	HOLD_INTERRUPTS();

	/*
	 * 在这里循环尝试在每次被LWLockRelease信号唤醒后获取锁。
	 *
	 * 注意：让LWLockRelease实际授予我们锁似乎更好，而不是重试并可能不得不回去睡觉。但是在实践中，这样做不好，因为这意味着每当两个或多个进程争用同一个锁时，每个锁获取都会发生进程交换。由于LWLocks通常用于保护不是很长的计算部分，因此一个进程需要能够在单个CPU时间片内多次获取和释放同一把锁，即使在存在竞争的情况下。能够做到这一点的效率超过了因为在一个释放的等待者最终运行时锁没有空而有时浪费一个进程调度周期的低效率。请参阅pgsql-hackers档案，日期为01年12月29日。
	 */
	for (;;)
	{
		bool		fc_mustwait;

		/*
		 * 第一次尝试获取锁，我们还没有/不再在等待队列中。
		 */
		fc_mustwait = fc_LWLockAttemptLock(fc_lock, fc_mode);

		if (!fc_mustwait)
		{
			fc_LOG_LWDEBUG("LWLockAcquire", fc_lock, "immediately acquired lock");
			break;				/* 获取锁 */
		}

		/*
		 * 好吧，此时我们第一次尝试没有成功获取锁。我们不能简单地将自己排队到队列末尾并等待被唤醒，因为此时锁很可能早已被释放。相反，将我们添加到队列并再次尝试获取锁。如果我们成功了，我们需要撤销排队并感到高兴，否则我们重新检查锁。如果我们仍然无法获取它，我们知道其他锁定者在释放时会看到我们的队列条目，因为它们在我们检查锁之前就存在。
		 */

		/* 添加到队列 */
		fc_LWLockQueueSelf(fc_lock, fc_mode);

		/* 如果需要，我们现在可以保证被唤醒 */
		fc_mustwait = fc_LWLockAttemptLock(fc_lock, fc_mode);

		/* 好吧，第二次尝试获取了锁，需要撤销排队 */
		if (!fc_mustwait)
		{
			fc_LOG_LWDEBUG("LWLockAcquire", fc_lock, "acquired, undoing queue");

			fc_LWLockDequeueSelf(fc_lock);
			break;
		}

		/*
		 * 等待被唤醒。
		 *
		 * 我们被唤醒的原因可能与LWLockRelease信号无关。如果是这样，则循环回去再次等待。一旦我们获得了LWLock，重新增加收到的附加信号计数。
		 */
		fc_LOG_LWDEBUG("LWLockAcquire", fc_lock, "waiting");

#ifdef LWLOCK_STATS
		fc_lwstats->block_count++;
#endif

		fc_LWLockReportWaitStart(fc_lock);
		if (TRACE_POSTGRESQL_LWLOCK_WAIT_START_ENABLED())
			TRACE_POSTGRESQL_LWLOCK_WAIT_START(T_NAME(fc_lock), fc_mode);

		for (;;)
		{
			PGSemaphoreLock(fc_proc->sem);
			if (fc_proc->lwWaiting == LW_WS_NOT_WAITING)
				break;
			fc_extraWaits++;
		}

		/* 重试，允许LWLockRelease再次释放等待者。 */
		pg_atomic_fetch_or_u32(&fc_lock->state, LW_FLAG_RELEASE_OK);

#ifdef LOCK_DEBUG
		{
			/* 不再等待 */
			uint32		fc_nwaiters PG_USED_FOR_ASSERTS_ONLY = pg_atomic_fetch_sub_u32(&fc_lock->nwaiters, 1);

			Assert(fc_nwaiters < MAX_BACKENDS);
		}
#endif

		if (TRACE_POSTGRESQL_LWLOCK_WAIT_DONE_ENABLED())
			TRACE_POSTGRESQL_LWLOCK_WAIT_DONE(T_NAME(fc_lock), fc_mode);
		fc_LWLockReportWaitEnd();

		fc_LOG_LWDEBUG("LWLockAcquire", fc_lock, "awakened");

		/* 现在循环回去再次尝试获取锁。 */
		fc_result = false;
	}

	if (TRACE_POSTGRESQL_LWLOCK_ACQUIRE_ENABLED())
		TRACE_POSTGRESQL_LWLOCK_ACQUIRE(T_NAME(fc_lock), fc_mode);

	/* 将锁添加到此后台持有的锁列表中 */
	held_lwlocks[num_held_lwlocks].lock = fc_lock;
	held_lwlocks[num_held_lwlocks++].mode = fc_mode;

	/*
	 * 修复进程等待信号量的计数以适应任何吸收的唤醒。
	 */
	while (fc_extraWaits-- > 0)
		PGSemaphoreUnlock(fc_proc->sem);

	return fc_result;
}

/*
 * LWLockConditionalAcquire - 以指定模式获取轻量级锁
 *
 * 如果锁不可用，则返回false且没有副作用。
 *
 * 如果成功，取消/死亡中断在锁释放之前被延迟。
 */
bool LWLockConditionalAcquire(LWLock *fc_lock, LWLockMode fc_mode)
{
	bool		fc_mustwait;

	AssertArg(fc_mode == LW_SHARED || fc_mode == LW_EXCLUSIVE);

	fc_PRINT_LWDEBUG("LWLockConditionalAcquire", fc_lock, fc_mode);

	/* 确保我们将有空间记住锁 */
	if (num_held_lwlocks >= MAX_SIMUL_LWLOCKS)
		elog(ERROR, "too many LWLocks taken");

	/*
	 * 在退出受LWLock保护的代码部分之前锁定取消/死亡中断。这确保中断不会干扰对共享内存中数据结构的操作。
	 */
	HOLD_INTERRUPTS();

	/* 检查锁 */
	fc_mustwait = fc_LWLockAttemptLock(fc_lock, fc_mode);

	if (fc_mustwait)
	{
		/* 获取锁失败，因此释放中断保持 */
		RESUME_INTERRUPTS();

		fc_LOG_LWDEBUG("LWLockConditionalAcquire", fc_lock, "failed");
		if (TRACE_POSTGRESQL_LWLOCK_CONDACQUIRE_FAIL_ENABLED())
			TRACE_POSTGRESQL_LWLOCK_CONDACQUIRE_FAIL(T_NAME(fc_lock), fc_mode);
	}
	else
	{
		/* 将锁添加到此后台持有的锁列表中 */
		held_lwlocks[num_held_lwlocks].lock = fc_lock;
		held_lwlocks[num_held_lwlocks++].mode = fc_mode;
		if (TRACE_POSTGRESQL_LWLOCK_CONDACQUIRE_ENABLED())
			TRACE_POSTGRESQL_LWLOCK_CONDACQUIRE(T_NAME(fc_lock), fc_mode);
	}
	return !fc_mustwait;
}

/*
 * LWLockAcquireOrWait - 获取锁，或等待直到它可用
 *
 * 此函数的语义有点奇怪。如果锁当前可用，则以给定模式获取它，并返回true。如果锁不立即可用，则该函数等待直到其释放并返回false，但不获取锁。
 *
 * 这目前用于WALWriteLock：当后台刷新WAL，持有WALWriteLock时，它可以副作用地刷新许多其他后台的提交记录。那些其他后台需要等待直到刷新完成，但不再需要获取锁。它们只需唤醒，观察它们的记录已被刷新，并返回。
 */
bool LWLockAcquireOrWait(LWLock *fc_lock, LWLockMode fc_mode)
{
	PGPROC	   *fc_proc = MyProc;
	bool		fc_mustwait;
	int			fc_extraWaits = 0;
#ifdef LWLOCK_STATS
	lwlock_stats *fc_lwstats;

	fc_lwstats = get_lwlock_stats_entry(fc_lock);
#endif

	Assert(fc_mode == LW_SHARED || fc_mode == LW_EXCLUSIVE);

	fc_PRINT_LWDEBUG("LWLockAcquireOrWait", fc_lock, fc_mode);

	/* 确保我们将有空间记住锁 */
	if (num_held_lwlocks >= MAX_SIMUL_LWLOCKS)
		elog(ERROR, "too many LWLocks taken");

	/*
	 * 在退出受LWLock保护的代码部分之前锁定取消/死亡中断。这确保中断不会干扰对共享内存中数据结构的操作。
	 */
	HOLD_INTERRUPTS();

	/*
	 * 注意：我们几乎使用相同的锁获取协议两次连续调用LWLockAcquire()。有关详细信息，请查看其注释。
	 */
	fc_mustwait = fc_LWLockAttemptLock(fc_lock, fc_mode);

	if (fc_mustwait)
	{
		fc_LWLockQueueSelf(fc_lock, LW_WAIT_UNTIL_FREE);

		fc_mustwait = fc_LWLockAttemptLock(fc_lock, fc_mode);

		if (fc_mustwait)
		{
			/*
			 * 等待被唤醒。像LWLockAcquire一样，准备好应对虚假的唤醒。
			 */
			fc_LOG_LWDEBUG("LWLockAcquireOrWait", fc_lock, "waiting");

#ifdef LWLOCK_STATS
			fc_lwstats->block_count++;
#endif

			fc_LWLockReportWaitStart(fc_lock);
			if (TRACE_POSTGRESQL_LWLOCK_WAIT_START_ENABLED())
				TRACE_POSTGRESQL_LWLOCK_WAIT_START(T_NAME(fc_lock), fc_mode);

			for (;;)
			{
				PGSemaphoreLock(fc_proc->sem);
				if (fc_proc->lwWaiting == LW_WS_NOT_WAITING)
					break;
				fc_extraWaits++;
			}

#ifdef LOCK_DEBUG
			{
				/* 不再等待 */
				uint32		fc_nwaiters PG_USED_FOR_ASSERTS_ONLY = pg_atomic_fetch_sub_u32(&fc_lock->nwaiters, 1);

				Assert(fc_nwaiters < MAX_BACKENDS);
			}
#endif
			if (TRACE_POSTGRESQL_LWLOCK_WAIT_DONE_ENABLED())
				TRACE_POSTGRESQL_LWLOCK_WAIT_DONE(T_NAME(fc_lock), fc_mode);
			fc_LWLockReportWaitEnd();

			fc_LOG_LWDEBUG("LWLockAcquireOrWait", fc_lock, "awakened");
		}
		else
		{
			fc_LOG_LWDEBUG("LWLockAcquireOrWait", fc_lock, "acquired, undoing queue");

			/*
			 * 在第二次尝试中获得锁，撤销排队。我们需要将这视为成功获得锁，否则我们不会
			 * 必然会唤醒阻止获取锁的人员。
			 */
			fc_LWLockDequeueSelf(fc_lock);
		}
	}

	/*
	 * 修复进程等待信号量的计数以适应任何吸收的唤醒。
	 */
	while (fc_extraWaits-- > 0)
		PGSemaphoreUnlock(fc_proc->sem);

	if (fc_mustwait)
	{
		/* 获取锁失败，因此释放中断保持 */
		RESUME_INTERRUPTS();
		fc_LOG_LWDEBUG("LWLockAcquireOrWait", fc_lock, "failed");
		if (TRACE_POSTGRESQL_LWLOCK_ACQUIRE_OR_WAIT_FAIL_ENABLED())
			TRACE_POSTGRESQL_LWLOCK_ACQUIRE_OR_WAIT_FAIL(T_NAME(fc_lock), fc_mode);
	}
	else
	{
		fc_LOG_LWDEBUG("LWLockAcquireOrWait", fc_lock, "succeeded");
		/* 将锁添加到此后台持有的锁列表中 */
		held_lwlocks[num_held_lwlocks].lock = fc_lock;
		held_lwlocks[num_held_lwlocks++].mode = fc_mode;
		if (TRACE_POSTGRESQL_LWLOCK_ACQUIRE_OR_WAIT_ENABLED())
			TRACE_POSTGRESQL_LWLOCK_ACQUIRE_OR_WAIT(T_NAME(fc_lock), fc_mode);
	}

	return !fc_mustwait;
}

/*
 * 当前状态下的 lwlock 是否需要等待变量值更改？
 *
 * 如果我们不需要等待，并且这是因为变量的值已更改，则将当前值存储在 newval 中。
 *
 * 如果锁是空闲的，则 *result 设置为 true， 否则为 false。
 */
static bool fc_LWLockConflictsWithVar(LWLock *fc_lock,
					   uint64 *fc_valptr, uint64 fc_oldval, uint64 *fc_newval,
					   bool *fc_result)
{
	bool		fc_mustwait;
	uint64		fc_value;

	/*
	 * 首先测试插槽是否现在是空闲的。
	 *
	 * XXX：在此之前，调用者使用了自旋锁，因此就当前使用而言，我们不需要这里的内存障碍。 
	 * 但这在一般情况下可能不安全。
	 */
	fc_mustwait = (pg_atomic_read_u32(&fc_lock->state) & LW_VAL_EXCLUSIVE) != 0;

	if (!fc_mustwait)
	{
		*fc_result = true;
		return false;
	}

	*fc_result = false;

	/*
	 * 使用 lwlock 的等待列表锁读取值，因为我们通常无法依赖原子 64 位读取/写入。 
	 * TODO：在支持原子 64 位读取/写入的平台上，自旋锁应优化掉。
	 */
	fc_LWLockWaitListLock(fc_lock);
	fc_value = *fc_valptr;
	fc_LWLockWaitListUnlock(fc_lock);

	if (fc_value != fc_oldval)
	{
		fc_mustwait = false;
		*fc_newval = fc_value;
	}
	else
	{
		fc_mustwait = true;
	}

	return fc_mustwait;
}

/*
 * LWLockWaitForVar - 等待直到锁空闲，或变量被更新。
 *
 * 如果锁被持有且 *valptr 等于 oldval，则等待直到锁被释放，或者锁持有者通过调用
 * LWLockUpdateVar 更新 *valptr。如果退出时锁是空闲的（立即或等待后），则返回 true。 
 * 如果锁仍然被持有，但 *valptr 不再与 oldval 匹配，则返回 false 并将 *newval 设置为 *valptr 中的当前值。
 *
 * 注意：此函数忽略共享锁持有者；如果锁以共享模式被持有，则返回 'true'。
 */
bool LWLockWaitForVar(LWLock *fc_lock, uint64 *fc_valptr, uint64 fc_oldval, uint64 *fc_newval)
{
	PGPROC	   *fc_proc = MyProc;
	int			fc_extraWaits = 0;
	bool		fc_result = false;
#ifdef LWLOCK_STATS
	lwlock_stats *fc_lwstats;

	fc_lwstats = get_lwlock_stats_entry(fc_lock);
#endif

	fc_PRINT_LWDEBUG("LWLockWaitForVar", fc_lock, LW_WAIT_UNTIL_FREE);

	/*
	 * 在我们在锁上休眠时锁定取消/终止中断。 
	 * 如果我们被中断，则没有清理机制可以将我们从等待队列中移除。
	 */
	HOLD_INTERRUPTS();

	/*
	 * 在每次被信号后循环检查锁的状态。
	 */
	for (;;)
	{
		bool		fc_mustwait;

		fc_mustwait = fc_LWLockConflictsWithVar(fc_lock, fc_valptr, fc_oldval, fc_newval,
										  &fc_result);

		if (!fc_mustwait)
			break;				/* 锁是空闲的或值不匹配 */

		/*
		 * 将自己添加到等待队列。注意这可能是竞争条件，其他人
		 * 可能会在我们完成排队前唤醒。注意：我们使用了几乎
		 * 相同的两次连续获取锁协议，如
		 * LWLockAcquire()。有关详细信息，请查看其注释。唯一的
		 * 区别是，在检查锁的状态时，我们还必须检查变量的值。
		 */
		fc_LWLockQueueSelf(fc_lock, LW_WAIT_UNTIL_FREE);

		/*
		 * 设置 RELEASE_OK 标志，以确保我们在锁释放时立即被唤醒。
		 */
		pg_atomic_fetch_or_u32(&fc_lock->state, LW_FLAG_RELEASE_OK);

		/*
		 * 现在可以保证在必要时被唤醒。重新检查锁
		 * 和变量的状态。
		 */
		fc_mustwait = fc_LWLockConflictsWithVar(fc_lock, fc_valptr, fc_oldval, fc_newval,
										  &fc_result);

		/* 好的，排队后没有冲突。撤销排队。 */
		if (!fc_mustwait)
		{
			fc_LOG_LWDEBUG("LWLockWaitForVar", fc_lock, "free, undoing queue");

			fc_LWLockDequeueSelf(fc_lock);
			break;
		}

		/*
		 * 等待被唤醒。
		 *
		 * 我们被唤醒的原因可能与LWLockRelease信号无关。如果是这样，则循环回去再次等待。一旦我们获得了LWLock，重新增加收到的附加信号计数。
		 */
		fc_LOG_LWDEBUG("LWLockWaitForVar", fc_lock, "waiting");

#ifdef LWLOCK_STATS
		fc_lwstats->block_count++;
#endif

		fc_LWLockReportWaitStart(fc_lock);
		if (TRACE_POSTGRESQL_LWLOCK_WAIT_START_ENABLED())
			TRACE_POSTGRESQL_LWLOCK_WAIT_START(T_NAME(fc_lock), LW_EXCLUSIVE);

		for (;;)
		{
			PGSemaphoreLock(fc_proc->sem);
			if (fc_proc->lwWaiting == LW_WS_NOT_WAITING)
				break;
			fc_extraWaits++;
		}

#ifdef LOCK_DEBUG
		{
			/* 不再等待 */
			uint32		fc_nwaiters PG_USED_FOR_ASSERTS_ONLY = pg_atomic_fetch_sub_u32(&fc_lock->nwaiters, 1);

			Assert(fc_nwaiters < MAX_BACKENDS);
		}
#endif

		if (TRACE_POSTGRESQL_LWLOCK_WAIT_DONE_ENABLED())
			TRACE_POSTGRESQL_LWLOCK_WAIT_DONE(T_NAME(fc_lock), LW_EXCLUSIVE);
		fc_LWLockReportWaitEnd();

		fc_LOG_LWDEBUG("LWLockWaitForVar", fc_lock, "awakened");

		/* 现在循环回来再检查一次锁的状态。 */
	}

	/*
	 * 修复进程等待信号量的计数以适应任何吸收的唤醒。
	 */
	while (fc_extraWaits-- > 0)
		PGSemaphoreUnlock(fc_proc->sem);

	/*
	 * 现在可以允许取消/死亡中断。
	 */
	RESUME_INTERRUPTS();

	return fc_result;
}


/*
 * LWLockUpdateVar - 原子更新变量并唤醒等待者
 *
 * 将 *valptr 设置为 'val'，并唤醒所有等待我们的进程，使用
 * LWLockWaitForVar()。设置值和唤醒进程是原子性的，因此
 * 任何在同一锁上调用 LWLockWaitForVar() 的进程都会
 * 确保看到新值，并相应地采取行动。
 *
 * 调用者必须以独占模式保持锁。
 */
void LWLockUpdateVar(LWLock *fc_lock, uint64 *fc_valptr, uint64 fc_val)
{
	proclist_head fc_wakeup;
	proclist_mutable_iter fc_iter;

	fc_PRINT_LWDEBUG("LWLockUpdateVar", fc_lock, LW_EXCLUSIVE);

	proclist_init(&fc_wakeup);

	fc_LWLockWaitListLock(fc_lock);

	Assert(pg_atomic_read_u32(&fc_lock->state) & LW_VAL_EXCLUSIVE);

	/* 更新锁的值 */
	*fc_valptr = fc_val;

	/*
	 * 查看是否有 LW_WAIT_UNTIL_FREE 等待者需要被唤醒。
	 * 它们总是在队列的前面。
	 */
	proclist_foreach_modify(fc_iter, &fc_lock->waiters, lwWaitLink)
	{
		PGPROC	   *fc_waiter = GetPGProcByNumber(fc_iter.cur);

		if (fc_waiter->lwWaitMode != LW_WAIT_UNTIL_FREE)
			break;

		proclist_delete(&fc_lock->waiters, fc_iter.cur, lwWaitLink);
		proclist_push_tail(&fc_wakeup, fc_iter.cur, lwWaitLink);

		/* 请参见 LWLockWakeup() */
		Assert(fc_waiter->lwWaiting == LW_WS_WAITING);
		fc_waiter->lwWaiting = LW_WS_PENDING_WAKEUP;
	}

	/* 我们已经完成了对锁自身共享状态的更新。 */
	fc_LWLockWaitListUnlock(fc_lock);

	/*
	 * 唤醒我从队列中移除的任何等待者。
	 */
	proclist_foreach_modify(fc_iter, &fc_wakeup, lwWaitLink)
	{
		PGPROC	   *fc_waiter = GetPGProcByNumber(fc_iter.cur);

		proclist_delete(&fc_wakeup, fc_iter.cur, lwWaitLink);
		/* 查看 LWLockWakeup() 中关于此屏障的注释 */
		pg_write_barrier();
		fc_waiter->lwWaiting = LW_WS_NOT_WAITING;
		PGSemaphoreUnlock(fc_waiter->sem);
	}
}


/*
 * LWLockRelease - 释放之前获取的锁
 */
void LWLockRelease(LWLock *fc_lock)
{
	LWLockMode	fc_mode;
	uint32		fc_oldstate;
	bool		fc_check_waiters;
	int			fc_i;

	/*
	 * 从所持锁的列表中移除锁。通常，但不是总是，它将是
	 * 最新获取的锁；因此从数组后向前搜索。
	 */
	for (fc_i = num_held_lwlocks; --fc_i >= 0;)
		if (fc_lock == held_lwlocks[fc_i].lock)
			break;

	if (fc_i < 0)
#ifdef FDD //cppcheck
	{
#endif
		elog(ERROR, "lock %s is not held", T_NAME(fc_lock));
#ifdef FDD //cppcheck
		return;
	}
#endif
	fc_mode = held_lwlocks[fc_i].mode;

	num_held_lwlocks--;
	for (; fc_i < num_held_lwlocks; fc_i++)
		held_lwlocks[fc_i] = held_lwlocks[fc_i + 1];

	fc_PRINT_LWDEBUG("LWLockRelease", fc_lock, fc_mode);

	/*
	 * 释放我对锁的占有，在此之后它可以立即被其他人获取，
	 * 即使我们仍然需要唤醒其他等待者。
	 */
	if (fc_mode == LW_EXCLUSIVE)
		fc_oldstate = pg_atomic_sub_fetch_u32(&fc_lock->state, LW_VAL_EXCLUSIVE);
	else
		fc_oldstate = pg_atomic_sub_fetch_u32(&fc_lock->state, LW_VAL_SHARED);

	/* 其他人无法拥有那种锁 */
	Assert(!(fc_oldstate & LW_VAL_EXCLUSIVE));

	if (TRACE_POSTGRESQL_LWLOCK_RELEASE_ENABLED())
		TRACE_POSTGRESQL_LWLOCK_RELEASE(T_NAME(fc_lock));

	/*
	 * 我们仍在等待后端被调度，不要再次唤醒它们。
	 */
	if ((fc_oldstate & (LW_FLAG_HAS_WAITERS | LW_FLAG_RELEASE_OK)) ==
		(LW_FLAG_HAS_WAITERS | LW_FLAG_RELEASE_OK) &&
		(fc_oldstate & LW_LOCK_MASK) == 0)
		fc_check_waiters = true;
	else
		fc_check_waiters = false;

	/*
	 * 由于唤醒等待者需要获取自旋锁，仅在必要时执行此操作。
	 */
	if (fc_check_waiters)
	{
		/* XXX: 提交前删除？ */
		fc_LOG_LWDEBUG("LWLockRelease", fc_lock, "releasing waiters");
		fc_LWLockWakeup(fc_lock);
	}

	/*
	 * 现在可以允许取消/死亡中断。
	 */
	RESUME_INTERRUPTS();
}

/*
 * LWLockReleaseClearVar - 释放之前获取的锁，重置变量
 */
void LWLockReleaseClearVar(LWLock *fc_lock, uint64 *fc_valptr, uint64 fc_val)
{
	fc_LWLockWaitListLock(fc_lock);

	/*
	 * 在释放锁之前设置变量的值，这可以防止竞争条件，
	 * 其中新的锁定者获取锁，但尚未设置变量的值。
	 */
	*fc_valptr = fc_val;
	fc_LWLockWaitListUnlock(fc_lock);

	LWLockRelease(fc_lock);
}


/*
 * LWLockReleaseAll - 释放所有当前持有的锁
 *
 * 用于在 ereport(ERROR) 之后进行清理。这个函数和批量
 * LWLockRelease 调用之间的一个重要区别是，InterruptHoldoffCount
 * 在此操作中没有变化。这是必要的，因为 InterruptHoldoffCount
 * 已在错误恢复中设置为适当的水平。如果我们允许其在每个
 * 释放的锁上下降，则可能将其减少到零以下！
 */
void LWLockReleaseAll(void)
{
	while (num_held_lwlocks > 0)
	{
		HOLD_INTERRUPTS();		/* 匹配即将到来的 RESUME_INTERRUPTS */

		LWLockRelease(held_lwlocks[num_held_lwlocks - 1].lock);
	}
}


/*
 * LWLockHeldByMe - 测试我的进程是否以任何模式持有锁
 *
 * 这仅仅是作为调试支持。
 */
bool LWLockHeldByMe(LWLock *fc_l)
{
	int			fc_i;

	for (fc_i = 0; fc_i < num_held_lwlocks; fc_i++)
	{
		if (held_lwlocks[fc_i].lock == fc_l)
			return true;
	}
	return false;
}

/*
 * LWLockHeldByMe - 测试我的进程是否持有数组中的任何锁
 *
 * 这仅仅是作为调试支持。
 */
bool LWLockAnyHeldByMe(LWLock *fc_l, int fc_nlocks, size_t fc_stride)
{
	char	   *fc_held_lock_addr;
	char	   *fc_begin;
	char	   *fc_end;
	int			fc_i;

	fc_begin = (char *) fc_l;
	fc_end = fc_begin + fc_nlocks * fc_stride;
	for (fc_i = 0; fc_i < num_held_lwlocks; fc_i++)
	{
		fc_held_lock_addr = (char *) held_lwlocks[fc_i].lock;
		if (fc_held_lock_addr >= fc_begin &&
			fc_held_lock_addr < fc_end &&
			(fc_held_lock_addr - fc_begin) % fc_stride == 0)
			return true;
	}
	return false;
}

/*
 * LWLockHeldByMeInMode - 测试我的进程是否以给定模式持有锁
 *
 * 这仅仅是作为调试支持。
 */
bool LWLockHeldByMeInMode(LWLock *fc_l, LWLockMode fc_mode)
{
	int			fc_i;

	for (fc_i = 0; fc_i < num_held_lwlocks; fc_i++)
	{
		if (held_lwlocks[fc_i].lock == fc_l && held_lwlocks[fc_i].mode == fc_mode)
			return true;
	}
	return false;
}
